package org.fhm.zdte.common.service.channel;

import org.fhm.substrate.annotation.Setup;
import org.fhm.substrate.service.LoggerHandler;
import org.fhm.substrate.standard.ILogger;
import org.fhm.zdte.common.config.ProtocolConfiguration;
import org.fhm.zdte.common.constant.DataTransform;
import org.fhm.zdte.common.pojo.Address;
import org.fhm.zdte.common.protocol.AbstractMessageParser;
import org.fhm.zdte.common.protocol.AbstractMessageStandard;
import org.fhm.zdte.common.service.boot.AbstractStarter;
import org.fhm.zdte.common.standard.IChannelManager;
import org.fhm.zdte.common.task.ReceiveMessageDistributorTask;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.StandardSocketOptions;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * The abstract layer of manager for UDP-Channel.
 *
 * @author 谭波
 * @since 2023/10/14
 */
public abstract class AbstractUDPChannelManager implements IChannelManager {

    /**
     * The thread name of receive Packet-Data.
     */
    private final static String UDP_RECEIVE_THREAD = "udp-receive-thread";

    private final ILogger logger = LoggerHandler.getLogger(AbstractUDPChannelManager.class);
    /**
     * Channel close flag.
     */
    private final AtomicBoolean isClose = new AtomicBoolean(false);

    @Setup("->protocol.name")
    protected AbstractMessageParser abstractMessageParser;
    @Setup
    protected ProtocolConfiguration protocolConfiguration;
    /**
     * The interval between sending packets
     */
    protected int sleepTime;
    /**
     * Datagram-Channel.
     */
    protected DatagramChannel dc;
    @Setup
    private ReceiveMessageDistributorTask receiveMessageDistributorTask;
    @Setup
    private AbstractStarter abstractStarter;
    private Selector selector;

    @Override
    public IChannelManager openChannel(int port) {
        try {
            selector = Selector.open();
            dc = DatagramChannel.open();
            dc.configureBlocking(false);
            dc.setOption(StandardSocketOptions.SO_RCVBUF, protocolConfiguration.getPacketLength() * 1024);
            dc.register(selector, SelectionKey.OP_READ);
            dc.bind(new InetSocketAddress(port));
            logger.info("server bind port {}", port);
        } catch (IOException e) {
            closeChannel();
            logger.error(e, "server bind port fail");
        }
        return this;
    }

    @Override
    public void enableChannel() {
        new Thread(() -> enableUDPChannel(getBuffer()), UDP_RECEIVE_THREAD).start();
    }

    @Override
    public void sendMessage(AbstractMessageStandard message, Address isa) {
        try {
            byte[] bytes = abstractMessageParser.encoding(message);
            int length = bytes.length;
            if (length > protocolConfiguration.getPacketLength()) {
                logger.warn("the maximum sending capacity is exceeded, length: {} ; message: {}", length, message);
                return;
            }
            if (isClose.get())
                return;
            if (sleepTime != 0)
                TimeUnit.MICROSECONDS.sleep(sleepTime);
            ByteBuffer wrap = ByteBuffer.wrap(bytes);
            int port = isa.getPort();
            synchronized (this) {
                dc.send(wrap, new InetSocketAddress(isa.getIpv4(), port));
            }
        } catch (NullPointerException e) {
            logger.error("failed to send Packet-Data", e);
        } catch (ClosedChannelException e) {
            disposeChannelClosedException();
        } catch (IOException e) {
            logger.error(e, "failed to send Packet-Data");
        } catch (InterruptedException ignore) {
        }
    }

    @Override
    public void closeChannel() {
        if (Objects.nonNull(selector) && selector.isOpen()) {
            selector.wakeup();
            try {
                Thread.sleep(50);
                selector.close();
            } catch (IOException e) {
                logger.error("failed to shutdown selector", e);
            } catch (InterruptedException ignore) {
            }
        }
        if (Objects.nonNull(dc) && dc.isOpen()) {
            boolean curMark;
            do {
                curMark = isClose.get();
            } while (!isClose.compareAndSet(curMark, true));
            try {
                dc.close();
            } catch (IOException e) {
                logger.error("failed to shutdown DatagramChannel", e);
            }
        }
    }

    private void enableUDPChannel(ByteBuffer buffer) {
        try {
            while (!isClose.get()) {
                selector.select();
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    if (iterator.next().isReadable()) {
                        SocketAddress receive = dc.receive(buffer);
                        if (receive instanceof InetSocketAddress) {
                            InetSocketAddress receiveAddress = (InetSocketAddress) receive;
                            byte[] bytes = exportByteBuffer(buffer);
                            receiveMessageDistributorTask
                                    .getTaskAndSubmit(bytes, new Address(receiveAddress));
                        } else {
                            logger.warn(receive + " is illegal address");
                            break;
                        }
                    }
                    iterator.remove();
                }
            }
        } catch (ClosedChannelException e) {
            disposeChannelClosedException();
        } catch (IOException e) {
            closeChannel();
        }
    }

    private void disposeChannelClosedException() {
        logger.warn("DatagramChannel stealth shutdown exception");
        try {
            abstractStarter.close();
        } catch (Exception e) {
            logger.warn("fail to close program", e);
        }
    }

    /**
     * Obtain a standard-length {@link Buffer}.
     *
     * @return {@link Buffer}.
     */
    private ByteBuffer getBuffer() {
        return ByteBuffer.allocateDirect(protocolConfiguration.getPacketLength());
    }

    protected void setSleepTime() {
        sleepTime = DataTransform.NUMBER_OF_FLOWS;
        logger.info("each packet sends {} bytes and stays in {} microseconds", protocolConfiguration.getPacketLength(), sleepTime);
    }

    /**
     * Export Byte-Array from {@link Buffer}.
     *
     * @param buffer {@link Buffer}.
     * @return Byte-Array.
     */
    private byte[] exportByteBuffer(ByteBuffer buffer) {
        buffer.flip();
        int len = buffer.limit();
        byte[] res = new byte[len];
        for (int i = 0; i < len; i++) {
            res[i] = buffer.get(i);
        }
        buffer.clear();
        return res;
    }
}
